package server;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.Iterator;

import javax.sql.XAConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;

import org.postgresql.xa.PGXADataSource;

import protocol.ProtocolServer;
import protocol.SqlCommandType;
import server.tasks.UpdateTask;
import server.dbInfo.ColumnInfo;
import server.dbInfo.TableInfo;

public class SqlHandler {
	private static ProtocolServer proto = new ProtocolServer();
		
	public static SqlCommandType checkCommandType(String sql) throws SQLException {
		sql = sql.trim().toLowerCase();
		
		if(sql.startsWith("select"))
			return SqlCommandType.Query;
		else if(sql.startsWith("update") || sql.startsWith("insert") || sql.startsWith("delete"))
			return SqlCommandType.NonQuery;
		else
			throw new SQLException("Obslugiwane typy komend to select, insert, update, delete");
	}
		
//	public static SqlResult executeSql(String sql, Object lock, ConnectionProvider provider, ExecutorService exec) throws Throwable {
//		SqlResult result = new SqlResult();		
//		SqlCommandType commandType = checkCommandType(sql);
//
//		if(commandType == SqlCommandType.Query){
//			result.setValue(executeQuery(sql, provider));
//			result.setType(SqlCommandType.Query);
//		} else if  (commandType == SqlCommandType.NonQuery) {
//			synchronized(lock) { 	
//				result.setValue(executeNonQuery(sql, provider, exec));
//				result.setType(SqlCommandType.NonQuery);
	// Collection<NodeInfo> dissattachedNodes = provider.getDisattachedNodes();
//				for(NodeInfo node : dissattachedNodes){
//					node.pushMissedNonQuery(sql);
//				}
//			}
//		}		
//		
//		return result;
//	}
	
	//implementacja 2 phase commit na podstawie artykulu http://www.cs.put.poznan.pl/mzakrzewicz/pubs/ploug03a.pdf
	public static String executeNonQuery(String nonQuery, ConnectionProvider provider, ExecutorService exec) throws Throwable {		
		int updateCount = 0;
		ArrayList<UpdateTask> startedTasks = startUpdateTasks(nonQuery, provider.getAttachedNodes(), exec);

		boolean doCommit = true;
		
		//zbieramy wyniki
		try {
			for(UpdateTask t : startedTasks) {				
				t.future.get(); //czekamy do zakonczenia zadania
				
				if(!((t.PrepareResult == XAResource.XA_OK))) {
					doCommit = false; //jesli ktorykolwiek prepare sie nie udal to nie bedziemy commitowac 				
				}	
			}		
		} catch (Exception e) {
			rollbackTasks(startedTasks); //tu tez moze byc wyjatek
			throw e.getCause();
		}			
		
		boolean transactionInterrupted = false;
		
		for(UpdateTask t : startedTasks){
			try {
				XAResource resource = t.getResource();
				if (t.PrepareResult == XAResource.XA_OK && doCommit)
					resource.commit(t.getXid(), false); 	//false == 2 phase commit
				else if(t.PrepareResult == XAResource.XA_OK)
					resource.rollback(t.getXid());
			} catch (XAException e) {
				//z jakichs powodow nie udalo sie dokonczyc transakcji
				//trzeba oznaczyc wezel jako taki, ktory posiada wiszaca transakcje
				NodeInfo node = t.getNode();
				node.updateFailured(t.getXid(), 
									doCommit ? UpdateFailureType.Commit : UpdateFailureType.Rollback);
				transactionInterrupted = true;
				Logger.logEvent(e.getMessage(), Logger.LOG_ERROR);
			}
		}
		
		if(transactionInterrupted)
			throw new UpdateException("Transakcja nie zostala ukonczona na wszystkich wezlach");			
		
		updateCount = startedTasks.get(0).UpdateCount;
		
		for(UpdateTask t : startedTasks) {
			t.dispose();
		}
		
		return proto.formatNonQuery(updateCount);
	}
	
	private static ArrayList<UpdateTask> startUpdateTasks(String sql, Iterable<NodeInfo> nodes, ExecutorService exec) {
		ArrayList<UpdateTask> results = new ArrayList<UpdateTask>();
		byte i = 0;
		
		for(NodeInfo info : nodes) {
			UpdateXid xid = new UpdateXid(i); //Xid z kolejnym id brancha transakcji
			i++;
			UpdateTask task = new UpdateTask(xid, info, sql);
			results.add(task);
		}

		for(UpdateTask t : results){
			Future<UpdateTask> futureResult = exec.submit(t);	
			t.future = futureResult;
		}		
		
		return results; //wszystkie zadania uruchomione (zapisane do uruchomienia w egzekutorze)
	}
	
	private static void rollbackTasks(Collection<UpdateTask> startedTasks) throws UpdateException {
		boolean rollbackError = false;
		for(UpdateTask t : startedTasks){
			if(t.PrepareResult != null) { //to znaczy ze task zdolal wykonac prepare
				try {
					t.getResource().rollback(t.getXid());
				} catch (XAException e) {
					//z jakichs powodow nie udalo sie rollbackow transakcji
					//trzeba oznaczyc wezel jako taki, ktory posiada wiszaca transakcje
					NodeInfo node = t.getNode();
					node.updateFailured(t.getXid(), UpdateFailureType.Rollback);
					Logger.logEvent(e.getMessage(), Logger.LOG_ERROR);
					rollbackError = true;
				}
			}
			t.dispose();
		}
	
		if(rollbackError) {
			throw new UpdateException("Transakcja rozproszona zostala przerwana po fazie prepare");
		}
	}
	
	public static String executeQuery(String query, ConnectionProvider balancer) throws SQLException {	
		NodeInfo info = balancer.nextConnection();
		Connection connection = DriverManager.getConnection(info.getUrl(), info.getUser(), info.getPassword());
		Statement sqlStatement = connection.createStatement();					
		sqlStatement.execute(query);
		ResultSet resultData = sqlStatement.getResultSet();
		String result = proto.formatResultSet(resultData);
		connection.close();
		System.out.println();
		return result;
	}

	public static boolean finishTransactions(NodeInfo node)  {
		
		if(node.getFailedOperation() == null)
			return false;		
		
		try {
			PGXADataSource source = new PGXADataSource();
			prepareXASource(source, node);				
			XAConnection XConn = source.getXAConnection();
			XAResource XResource = XConn.getXAResource();	
			
			if(node.getFailedOperation() == UpdateFailureType.Commit)
				XResource.commit(node.getFailedXid(), false);	
			else if(node.getFailedOperation() == UpdateFailureType.Rollback)
				XResource.rollback(node.getFailedXid());
			node.updateRecovered();
		
		
			XConn.close();
		} catch (Exception e) {
			return false;
		}
		
		return true;
	}

	public static boolean canConnect(NodeInfo info) {
		try {
			Connection connection = DriverManager.getConnection(info.getUrl(),
					info.getUser(), info.getPassword());
	
			connection.getMetaData();
			connection.close();
		} catch (SQLException e) {
			return false;
		}
	
		return true;
	}		
	
	public static void prepareXASource(PGXADataSource source, NodeInfo node) {
		source.setDatabaseName(node.getDbName());
		source.setUser(node.getUser());
		source.setPassword(node.getPassword());
		source.setPortNumber(node.getPort());
		source.setServerName(node.getIP()); //ip to server name
											//http://www.java2s.com/Open-Source/Java-Document/Database-JDBC-Connection-Pool/postgresql/org/postgresql/ds/common/BaseDataSource.java.htm
	}	

	/*
	* @param source wezel zrodlowy
	* @target target wezel docelowy zrodlowy
	* returns liczba skopiowanych wierszy
	*/
	public static int copyNode(NodeInfo source, NodeInfo target) throws SQLException{
		
		//otworzyc polaczenie na wezle source i target
		Connection sourceConn = DriverManager.getConnection(source.getUrl(), source.getUser(), source.getPassword());
		Connection targetConn = DriverManager.getConnection(target.getUrl(), target.getUser(), target.getPassword());
		
		Statement sourceStmt = sourceConn.createStatement();
		Statement targetStmt = targetConn.createStatement();
			

		//otwarcie transakcji na wezle target 
		sourceConn.setAutoCommit(false);
		int inserts = 0;
		int n=0;
		
		try
		{
		//pobieram nazwe bazy danych do usuniecia i uswam ja z wezla target
		/*String q1 = new String("select t.constraint_name, t.table_name, t.constraint_type, c.table_name, " +
							" c.column_name from information_schema.table_constraints t," +
							" information_schema.constraint_column_usage c where t.constraint_name = c.constraint_name " +
							" and t.constraint_type = 'FOREIGN KEY' and c.table_name = 'mytable';");
		
		String query = new String("DROP DATABASE "+target.getDbName()+";");		
		//String query = new String("DROP DATABASE "+target.getDbName()+";");
		ResultSet res = targetStmt.executeQuery(query);
		
		*/
		String q = new String("SELECT COUNT(*), table_name FROM information_schema.tables group by table_name;");
		ResultSet r = sourceStmt.executeQuery(q);
		r.next();
		if (r.next())
			n =  r.getInt(1);
		System.out.println(n);

		while (n>0)
		{
			--n;
			ArrayList<TableInfo> info = TableInfo.load(source);
						
			TableInfo colInf = info.iterator().next();
			String createTableQuery = new String ("CREATE TABLE "+colInf.tableName+" (");
			String nullable=new String("");
				
			ArrayList<ColumnInfo> column = colInf.columns;
			Iterator<ColumnInfo> it = column.iterator();
			while (it.hasNext()){
				ColumnInfo temp = it.next();
				
				if (!temp.isNullable) {
					nullable = "NOT NULL";
				}
				createTableQuery += temp.ColumnName + " " 
									+ temp.TypeName //+ 
									//" (" + temp.Precision +")";
									+ nullable;
				if (it.hasNext())
					createTableQuery +=	", ";
				else
					createTableQuery +=	");";
			}
			System.out.println(createTableQuery);
			//wykonanie creata
			ResultSet createTableResult = targetStmt.executeQuery(createTableQuery);
						
			server.dbInfo.KeyInfo key = new server.dbInfo.KeyInfo();
			String keyQuery = new String ("ALTER TABLE "+colInf.tableName+" ADD primary key ("+ key.Name+");"); 
			ResultSet keyResult = targetStmt.executeQuery(keyQuery);
			
			//inserty do tejze tabeli
			String selectTableQuery = new String ("SELECT * FROM "+colInf.tableName+";");
			System.out.println(selectTableQuery);
			ResultSet selectTableResult = sourceStmt.executeQuery(selectTableQuery);
			while ( selectTableResult.next() )
			{
				String insertQuery = new String ("INSERT INTO "+colInf.tableName+" VALUES (");
				for (int i=1;i<=colInf.columns.size();++i){
					insertQuery += selectTableResult.getInt(i);
					if (i == colInf.columns.size()-1)
						insertQuery += ", ";
					else
						insertQuery += ");";
				}
				ResultSet iserteResult = sourceStmt.executeQuery(insertQuery);
				//targetStmt.addBatch(insertQuery);
				
				System.out.println(insertQuery);
				//wyciagnac z batcha ile sie udalo insertow wykonac
				++inserts;
			}
			//targetStmt.executeBatch();
		}
		targetConn.commit();
		
		} catch(SQLException e) {
			targetConn.rollback();
			throw e; //wyjatek bedzie jeszcze obslugiwany "u gory"
		}
		finally
		{
			sourceConn.close();
			targetConn.close();
		}
		return inserts;
	}
	
	
}
